package com.ruoze.bdp.function.impl;

import com.alibaba.fastjson.JSON;
import com.ruoze.bdp.entity.TableStructVo;
import com.ruoze.bdp.entity.datasource.HBaseDataSourceParamDTO;
import com.ruoze.bdp.enums.DbType;
import com.ruoze.bdp.function.DataSourceFunction;
import com.ruoze.bdp.utils.HBaseUtil;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Table;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.stream.Collectors;

@Service
public class HBaseDataSourceFunction implements DataSourceFunction {
    @Override
    public boolean connectTest(String info) {
        try{
            HBaseDataSourceParamDTO dto = JSON.parseObject(info, HBaseDataSourceParamDTO.class);
            return HBaseUtil.testConnection(dto.getZkHost(), dto.getZkPort());
        }catch (Exception e){
            return false;
        }
    }

    @Override
    public List<String> getMetaDatabases(String info) {
        try {
            HBaseDataSourceParamDTO dto = JSON.parseObject(info, HBaseDataSourceParamDTO.class);
            Optional<Connection> connectionOpt = HBaseUtil.getHbaseConnection(dto.getZkHost(), dto.getZkPort());
            if(connectionOpt.isPresent()){
                Connection connection = connectionOpt.get();
                return HBaseUtil.getNamespaces(connection);
            }else{
                return null;
            }
        } catch (Exception e) {
            return null;
        }
    }

    @Override
    public List<String> getMetaTables(String info, String database) {
        try {
            HBaseDataSourceParamDTO dto = JSON.parseObject(info, HBaseDataSourceParamDTO.class);
            Optional<Connection> connectionOpt = HBaseUtil.getHbaseConnection(dto.getZkHost(), dto.getZkPort());
            if(connectionOpt.isPresent()){
                Connection connection = connectionOpt.get();
                return HBaseUtil.getTables(connection, database);
            }else{
                return null;
            }
        } catch (Exception e) {
            return null;
        }
    }

    @Override
    public List<Map<String, String>> getMetaColumns(String info, String database, String table) {
        try {
            HBaseDataSourceParamDTO dto = JSON.parseObject(info, HBaseDataSourceParamDTO.class);
            Optional<Connection> connectionOpt = HBaseUtil.getHbaseConnection(dto.getZkHost(), dto.getZkPort());
            if(connectionOpt.isPresent()){
                Connection connection = connectionOpt.get();
                return HBaseUtil.getFamilyColumns(connection, database, table)
                        .stream()
                        .map(i -> {
                            HashMap<String, String> columnMap = new HashMap<>(1);
                            columnMap.put(i, "cf");
                            return columnMap;
                        }).collect(Collectors.toList());
            }else{
                return null;
            }
        } catch (Exception e) {
            return null;
        }
    }

    @Override
    public boolean createDatabase(String info, String name) {
        try {
            HBaseDataSourceParamDTO dto = JSON.parseObject(info, HBaseDataSourceParamDTO.class);
            Optional<Connection> connectionOpt = HBaseUtil.getHbaseConnection(dto.getZkHost(), dto.getZkPort());
            if(connectionOpt.isPresent()){
                return HBaseUtil.createNamespace(connectionOpt.get(), name);
            }else{
                return false;
            }
        } catch (Exception e) {
            return false;
        }
    }

    @Override
    public DbType getSourceType() {
        return DbType.HBASE;
    }

    @Override
    public boolean createTable(String info, TableStructVo tableVo) {
        try {
            HBaseDataSourceParamDTO dto = JSON.parseObject(info, HBaseDataSourceParamDTO.class);
            Optional<Connection> connectionOpt = HBaseUtil.getHbaseConnection(dto.getZkHost(), dto.getZkPort());
            if(connectionOpt.isPresent()){
                String namespace = tableVo.getDatabaseName();
                String tableName = tableVo.getTableName();
                List<String> cfs = tableVo.getColumnDefinitionList()
                        .stream()
                        .map(Map::keySet)
                        .map(i -> i.iterator().next())
                        .collect(Collectors.toList());
                return HBaseUtil.createTable(connectionOpt.get(), namespace, tableName, cfs);
            }else{
                return false;
            }
        } catch (Exception e) {
            return false;
        }
    }
}
